package com.kancy.stream.rabbit.config;

import com.kancy.stream.rabbit.properties.ChannelProperties;
import com.kancy.stream.rabbit.properties.RabbitMessageProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.CollectionUtils;

/**
 * RabbitMessageChannelListenerRegistrar
 *
 * @author kancy
 * @date 2020/3/8 0:55
 */
public class RabbitMessageChannelListenerRegistrar implements InitializingBean {
    private static Logger log = LoggerFactory.getLogger(RabbitMessageChannelListenerRegistrar.class);

    private final RabbitListenerContainerFactory rabbitListenerContainerFactory;
    private final RabbitListenerEndpointRegistry rabbitListenerEndpointRegistry;
    private final DynamicInterfaceBeanResolver dynamicInterfaceBeanResolver;
    private final BeanFactory beanFactory;
    private final RabbitMessageProperties rabbitMessageProperties;

    public RabbitMessageChannelListenerRegistrar(RabbitListenerContainerFactory rabbitListenerContainerFactory,
                                                 RabbitListenerEndpointRegistry rabbitListenerEndpointRegistry,
                                                 DynamicInterfaceBeanResolver dynamicInterfaceBeanResolver,
                                                 BeanFactory beanFactory,
                                                 RabbitMessageProperties rabbitMessageProperties) {
        this.rabbitListenerContainerFactory = rabbitListenerContainerFactory;
        this.rabbitListenerEndpointRegistry = rabbitListenerEndpointRegistry;
        this.dynamicInterfaceBeanResolver = dynamicInterfaceBeanResolver;
        this.beanFactory = beanFactory;
        this.rabbitMessageProperties = rabbitMessageProperties;
    }

    @Override
    public void afterPropertiesSet() {
        rabbitMessageProperties.getChannel().entrySet().stream().forEach(e ->{
            ChannelProperties channelProperties = e.getValue();
            if (!CollectionUtils.isEmpty(channelProperties.getConsumer())){
                rabbitListenerEndpointRegistry.registerListenerContainer(getRabbitListenerEndpoint(e.getKey()), rabbitListenerContainerFactory);
                log.debug("Register rabbit listener container [{}] success.", e.getKey());
            }
        });
    }

    /**
     * getRabbitListenerEndpoint
     * @param channelKey
     * @return
     */
    private RabbitMessageChannelListenerEndpoint getRabbitListenerEndpoint(String channelKey) {
        return new RabbitMessageChannelListenerEndpoint(rabbitMessageProperties, channelKey, beanFactory, dynamicInterfaceBeanResolver);
    }

}
